package com.sun.clients.runner;

import com.sun.clients.kafka.ConsumerExecutor;
import com.sun.clients.properties.AppProperty;
import com.sun.clients.service.MessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class KafkaStartRunner implements CommandLineRunner {

    private final KafkaConsumer<String, String> kafkaConsumer;
    private final ThreadPoolExecutor threadPoolExecutor;
    private final MessageService messageService;
    private final AppProperty appProperty;

    @Override
    public void run(String... args) {
        ConsumerExecutor consumerExecutor = new ConsumerExecutor(kafkaConsumer,
                threadPoolExecutor, messageService, appProperty);
        consumerExecutor.execute();
        consumerExecutor.shutdown();
    }

}
